package cn.tedu.micro.demo.demo03;

import cn.tedu.micro.demo.client.RabbitChannelManager;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.TimeoutException;

@Slf4j
public class RabbitDemo03 {
    //生命一个队列组件
    @Test
    public void declare() throws IOException {
        Channel channel = RabbitChannelManager.newChannel();
        channel.queueDeclare(
                "queue01",
                false,
                false,
                false,
                null);
    }
    @Test
    public void send() throws IOException, InterruptedException, TimeoutException {
        Channel channel = RabbitChannelManager.newChannel();
        byte[] body="hello rabbitmq".getBytes(StandardCharsets.UTF_8);

        //发送之前开启同步确认发送
        channel.confirmSelect();
        //在发送客户端绑定一个异步监听对象
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                log.info("消息发送成功");
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                log.info("消息发送失败");
            }
        });
        long start=System.currentTimeMillis();
        channel.basicPublish("","queue01",null,body);
        long end=System.currentTimeMillis();
        log.info("消息发送结束,消耗时间:{}",end-start);
        Thread.sleep(1000);
    }
    //测试消费监听
    @Test
    public void consume() throws IOException, InterruptedException {
        //1.拿到客户端对象
        Channel channel = RabbitChannelManager.newChannel();
        //2.消费方法
        //2.1构建消费方法的入参 delivery 回调(messageArrived类似)
        DeliverCallback deliver=new DeliverCallback() {
            /**
             * 方法作用: 只要监听队列 有消息,就会给当前这个消费者推送消息
             * 每条消息,在消费者中handle处理
             * String consumerTag: 消费者身份 rabbitmq分配的表示tag
             * Delivery message: 就是消息对象 发了什么 这里就能收到什么
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //底层消费入口 获取消息 解析消息,调用业务完成消费
                //1.消息体 body
                byte[] body = message.getBody();
                String msg=new String(body,StandardCharsets.UTF_8);
                //2.消息属性
                AMQP.BasicProperties properties = message.getProperties();
                String expiration = properties.getExpiration();
                Integer priority = properties.getPriority();
                Map<String, Object> headers = properties.getHeaders();
                headers.get("name");
                headers.get("age");
                //3.获取消息途径过程 交换机 routingKey
                String exchange = message.getEnvelope().getExchange();
                String routingKey = message.getEnvelope().getRoutingKey();
                log.info("消息:{},从交换机:{},使用路由key:{}达到消费者",msg,exchange,routingKey);
            }
        };
        //2.2构造消费方法的入参 cancel 回调
        CancelCallback cancel=new CancelCallback(){
            /**
             *方法作用: 消息费拒绝 取消消费的时候 调用cancel
             *String consumerTag: 消费者身份标识
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                log.info("消费者:{},取消了消费",consumerTag);
            }
        };
        channel.basicConsume("queue01",true,
                deliver,
                cancel);
        Thread.sleep(10000000);
    }
}
